package com.pai4j.connect.messagequeue.consumer.service;
import com.pai4j.common.enums.SubjectRelationTypeEnum;
import com.pai4j.common.enums.messagequeue.MessageQueueEnum;
import com.pai4j.common.bean.PAIResponseBean;
import com.pai4j.common.bean.PAIResponseBeanUtil;
import com.pai4j.common.constants.ConnectConstants;
import com.pai4j.common.enums.MessageTypeEnum;
import com.pai4j.common.service.messagequeue.producer.MessageQueueProducer;
import com.pai4j.common.util.JsonUtil;
import com.pai4j.connect.websocket.handler.ChatWebSocketEndPointServletHandler;
import com.pai4j.domain.dto.queue.ArticleDelayMessageDTO;
import com.pai4j.domain.vo.request.mess.ChatMessageRequestVO;
import com.pai4j.domain.vo.response.mess.ChatMessageResponseVO;
import com.pai4j.domain.vo.response.mess.SubjectAccountArticleMessageVO;
import com.pai4j.domain.vo.response.subjectaccount.ArticleBaseResponseVO;
import com.pai4j.remote.pgc.SubjectArticleClient;
import com.pai4j.remote.pgc.SubjectRelationClient;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;


/**
 * 专题推文消费服务
 *
 * @author: CYM-pai
 * @date: 2025/06/05 11:31
 **/
@Slf4j
@Service
public class ArticlePushConsumeService {

   private final Integer relationType = SubjectRelationTypeEnum.USER_SUBSCRIBE_SUBJECT.getType();
    @Resource
    private SubjectRelationClient subjectRelationClient;

    @Autowired
    private SubjectArticleClient subjectArticleClient;

    @Autowired
    private MessageQueueProducer<ChatMessageRequestVO, Long> messageQueueProducer;

    public void consume(String message) {

        /**
         * 发序列化解析消息对象
         */
        //将输入的 JSON 字符串 message 反序列化为一个 Set<String> 集合，即把 JSON 数组解析成多个字符串消息。
        Set<String> messages = JsonUtil.fromJson(message, Set.class);
        Set<ArticleDelayMessageDTO> messageDtos =
                messages.stream().map(mess ->
                        JsonUtil.fromJson(mess, ArticleDelayMessageDTO.class))
                        .collect(Collectors.toSet());
        /**
         * 消息推送:for循环遍历Set集合中的每一个ArticleDelayMessageDTO 对象
         */
        for (ArticleDelayMessageDTO eventMessage : messageDtos) {
            this.doPush(eventMessage);
        }
    }

    private void doPush(ArticleDelayMessageDTO eventMessage) {
        //获取公众号id
        Long subjectId =Long.valueOf(eventMessage.getSubjectAccount());
        //获取当前要推送的文章id
        Long articleId = eventMessage.getArticleId();
        // 获取文章详情，走feign 远程获取
        SubjectAccountArticleMessageVO articleVO = this.getArticleBaseFromRemote(articleId);
        if (articleVO == null) {
            return;
        }
        // 获取专题所有订阅用户
        Set<String> userIds = subjectRelationClient.getSubjectAllSubscriber(relationType,subjectId).getData();
        if (CollectionUtils.isEmpty(userIds)) {
            return;
        }
        ChatMessageResponseVO chatMessageResponseVO = new ChatMessageResponseVO();
        chatMessageResponseVO.setSubjectAccountArticle(articleVO);
        chatMessageResponseVO.setSenderId(String.valueOf(subjectId));
        chatMessageResponseVO.setType(MessageTypeEnum.SUBJECT_ACCOUNT_ARTICLE.getType());

        Set<String> bizCodes = ConnectConstants.ConnectBiz.getBizCode(ConnectConstants.BizEnum.CHAT);
        for (String bizCode : bizCodes) {
            Map<String, WebSocketSession> userWsMap = ChatWebSocketEndPointServletHandler.getSessions(bizCode, userIds);
            for (String userId : userIds) {
                WebSocketSession wsSession = userWsMap.get(userId);
                if (wsSession == null || !wsSession.isOpen()) {
                    continue;
                }
                chatMessageResponseVO.setReceiverId(userId);
                try {
                    /**
                     * 公众号文章推送
                     */
                    wsSession.sendMessage(new TextMessage(JsonUtil.toJsonString(chatMessageResponseVO)));
                    /**
                     *  推送数据持久化
                     */
                    this.persistentArticlePushMessage(String.valueOf(subjectId), userId, articleId);
                } catch (IOException e) {
                    // 不能抛异常出去，不能阻塞其他用户的推送
                    log.error("公众号文章推送异常 =====> publicAccount:{}, userId:{}, articleId:{}",
                            subjectId, userId, articleId, e);
                }

            }
        }
    }


    /**
     * 公众号推送消息持久化
     *
     */
    private void persistentArticlePushMessage(String subjectId, String userId, Long articleId) {

        ChatMessageRequestVO persistentMessage = new ChatMessageRequestVO();
        persistentMessage.setType(MessageTypeEnum.SUBJECT_ACCOUNT_ARTICLE.getType());
        persistentMessage.setSenderId(subjectId);
        persistentMessage.setReceiverId(userId);
        persistentMessage.setMessage(String.valueOf(articleId));
        /**
         * 走统一的消息持久化处理服务（服用了对话场景持久化服务）
         *
         */
        messageQueueProducer.send(MessageQueueEnum.QUEUE_PERSISTENT_CHAT_MESSAGE, persistentMessage);
    }

    /**
     * 获取文章详情，这里不需要返回正文（降低相应网络传输包大小，节约流量成本，提高接口性能）
     *
     * @param id
     * @return
     */
    private SubjectAccountArticleMessageVO getArticleBaseFromRemote(Long id) {
        PAIResponseBean<ArticleBaseResponseVO> responseBean = subjectArticleClient.detail(id, false);
        if (PAIResponseBeanUtil.isOk(responseBean)) {
            ArticleBaseResponseVO baseResponseVO = responseBean.getData();
            SubjectAccountArticleMessageVO publicAccountArticleMessageVO = new SubjectAccountArticleMessageVO();
            publicAccountArticleMessageVO.setArticleId(baseResponseVO.getId());
            publicAccountArticleMessageVO.setTime(baseResponseVO.getPlanPushTime());
            publicAccountArticleMessageVO.setTitle(baseResponseVO.getTitle());
            publicAccountArticleMessageVO.setCover(baseResponseVO.getCover());
            publicAccountArticleMessageVO.setDescription(baseResponseVO.getDescription());
            publicAccountArticleMessageVO.setRedirectUrl(baseResponseVO.getRedirectUrl());
            return publicAccountArticleMessageVO;
        }
        log.error("远程获取公众号文章详情失败 =====> id:{}, responseBean:{}", id, JsonUtil.toJsonString(responseBean));
        return null;
    }

}
